#pragma once

#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "index/ix.h"
#include "system/sm.h"

class IndexScanExecutor : public AbstractExecutor {
   private:
    std::string tab_name_;
    std::vector<Condition> conds_;
    RmFileHandle *fh_;
    std::vector<ColMeta> cols_;
    size_t len_;
    std::vector<Condition> fed_conds_;

    int index_no_;

    Rid rid_;
    std::unique_ptr<RecScan> scan_;

    SmManager *sm_manager_;

   public:
    IndexScanExecutor(SmManager *sm_manager, std::string tab_name, std::vector<Condition> conds, int index_no,
                      Context *context) {
        // lab3 task2 todo
        // 参考seqscan作法,实现indexscan构造方法
        // lab3 task2 todo

        // 在数据库管理系统中初始化一个索引扫描执行器。这个执行器用于基于索引和给定条件对表中的数据进行扫描。
        // 用于初始化一个索引扫描执行器
        index_no_ = index_no;// 设置索引编号
        rid_ = {-1, -1};// 初始化记录标识符为无效值
        scan_ = nullptr;// 初始化扫描器指针为nullptr
        sm_manager_ = sm_manager;// 保存数据库管理器的指针
        tab_name_ = std::move(tab_name);// 保存表名
        conds_ = std::move(conds);// 保存条件向量
        TabMeta &tab = sm_manager_->db_.get_table(tab_name_);// 获取表的元数据
        fh_ = sm_manager_->fhs_.at(tab_name_).get();// 获取与表相关的文件句柄
        cols_ = tab.cols;// 保存列的元数据
        len_ = cols_.back().offset + cols_.back().len;// 计算记录长度
        context_ = context;// 保存上下文指针
        // 创建一个映射，用于在条件比较操作中交换操作数
        std::map<CompOp, CompOp> swap_op =
        {
            {OP_EQ, OP_EQ}, {OP_NE, OP_NE}, {OP_LT, OP_GT}, {OP_GT, OP_LT}, {OP_LE, OP_GE}, {OP_GE, OP_LE},
        };
        for (auto &cond : conds_)
        {
            if (cond.lhs_col.tab_name != tab_name_)
            {
                // 如果条件的左侧列不在当前表中，则右侧列必须在当前表中
                assert(!cond.is_rhs_val && cond.rhs_col.tab_name == tab_name_);
                // 交换条件的左侧和右侧列
                std::swap(cond.lhs_col, cond.rhs_col);
                cond.op = swap_op.at(cond.op);
            }
        }
        fed_conds_ = conds_;
    }

    std::string getType() { return "indexScan"; }

    void beginTuple() {
        // 负责根据给定的条件初始化索引扫描，并开始迭代以找到满足这些条件的第一个记录
        check_runtime_conds();// 检查运行时条件，确保所有条件都满足

        // 索引可用，扫描索引
        auto ih = sm_manager_->ihs_.at(sm_manager_->get_ix_manager()->get_index_name(tab_name_, index_no_)).get();
        // 初始化索引的下界和上界
        Iid lower = ih->leaf_begin();
        Iid upper = ih->leaf_end();
        auto &index_col = cols_[index_no_];// 获取索引列的元数据
        for (auto &cond : fed_conds_) {
            if (cond.is_rhs_val && cond.op != OP_NE && cond.lhs_col.col_name == index_col.name) {
                char *rhs_key = cond.rhs_val.raw->data;// 获取条件中的右侧键值
                if (cond.op == OP_EQ) {// 等于操作符
                    // 设置下界和上界为等于rhs_key的范围
                    lower = ih->lower_bound(rhs_key);
                    upper = ih->upper_bound(rhs_key);
                }

                // lab3 task2 todo
                /**
                 * else if(cond.op == ?){
                 *
                 * }else if(){
                 * ...
                 * }else if(){
                 * ...
                 * }
                 * ...
                 * else{
                 *  throw InternalError("Unexpected op type");
                 * }
                 *
                 *
                 */
                // 利用cond 进行索引扫描
                // lab3 task2 todo end
                else if (cond.op == OP_LT)// 小于操作符，设置上界
                {
                    upper = ih->lower_bound(rhs_key);
                }
                else if (cond.op == OP_LE)// 小于等于操作符，设置上界
                {
                    upper = ih->upper_bound(rhs_key);
                }
                else if (cond.op == OP_GT)// 大于操作符，设置下界
                {
                    lower = ih->upper_bound(rhs_key);
                }
                else if (cond.op == OP_GE)// 大于等于操作符，设置下界
                {
                    lower = ih->lower_bound(rhs_key);
                }
                else// 如果遇到未知的操作符，抛出内部错误
                {
                    throw InternalError("Unexpected op type");
                }
                break;
            }
        }
        scan_ = std::make_unique<IxScan>(ih, lower, upper, sm_manager_->get_bpm());
        // 获取第一条记录
        while (!scan_->is_end()) {
            rid_ = scan_->rid();// 获取当前记录的RID
            auto rec = fh_->get_record(rid_, context_);// 获取记录内容
            if (eval_conds(cols_, fed_conds_, rec.get())) {// 如果记录满足所有条件，则结束循环
                break;
            }
            scan_->next();// 移动到下一条记录
        }
    }

    void nextTuple() {
        // 负责在索引扫描中移动到下一个满足给定条件的记录
        check_runtime_conds();// 检查运行时条件，确保所有条件都满足
        assert(!is_end());// 断言当前不在扫描的末尾
        // lab3 task2 todo
        // 扫描到下一个满足条件的记录,赋rid_,中止循环
        // lab3 task2 todo end
        while (!scan_->is_end())// 当扫描没有到达末尾时，继续进行
        {
            scan_->next();// 移动到下一个记录
            if (scan_->is_end())// 如果移动后到达了扫描的末尾，则退出循环
            {
                break;
            }
            rid_ = scan_->rid();// 获取当前记录的RID（记录标识符）
            auto rec = fh_->get_record(rid_, context_);// 获取当前记录的内容
            if (eval_conds(cols_, fed_conds_, rec.get()))// 如果当前记录满足所有条件，则结束循环
            {
                break;
            }
        }
    }

    bool is_end() const override { return scan_->is_end(); }

    size_t tupleLen() const override { return len_; }

    const std::vector<ColMeta> &cols() const override { return cols_; }

    std::unique_ptr<RmRecord> Next() override {
        assert(!is_end());
        return fh_->get_record(rid_, context_);
    }

    void feed(const std::map<TabCol, Value> &feed_dict) override {
        // 用于在数据库查询执行器（如索引扫描器或顺序扫描器）中处理条件的动态部分。
        // 接收一个映射（feed_dict），该映射包含表列到值的映射，用于更新查询条件中引用的列。
        fed_conds_ = conds_;// 复制原始条件到fed_conds_，以便进行修改
        for (auto &cond : fed_conds_) {
            // lab3 task2 todo
            // 参考seqscan
            // lab3 task2 todo end
            if (cond.is_rhs_val)// 如果条件的右侧已经是一个值（而非列引用），则跳过
            {
                continue;
            }
            auto it = feed_dict.find(cond.rhs_col);// 在feed_dict中查找与条件右侧列相对应的值
            if (it != feed_dict.end())// 如果找到了对应的值
            {
                cond.rhs_val = it->second;// 更新条件的右侧为该值
                cond.is_rhs_val = true;// 标记条件的右侧现在是一个值
            }
        }
        check_runtime_conds();// 检查运行时条件，确保所有条件都满足
    }

    Rid &rid() override { return rid_; }

    void check_runtime_conds() {
        for (auto &cond : fed_conds_) {
            assert(cond.lhs_col.tab_name == tab_name_);
            if (!cond.is_rhs_val) {
                assert(cond.rhs_col.tab_name == tab_name_);
            }
        }
    }

    bool eval_cond(const std::vector<ColMeta> &rec_cols, const Condition &cond, const RmRecord *rec) {
        auto lhs_col = get_col(rec_cols, cond.lhs_col);
        char *lhs = rec->data + lhs_col->offset;
        char *rhs;
        ColType rhs_type;
        if (cond.is_rhs_val) {
            rhs_type = cond.rhs_val.type;
            rhs = cond.rhs_val.raw->data;
        } else {
            // rhs is a column
            auto rhs_col = get_col(rec_cols, cond.rhs_col);
            rhs_type = rhs_col->type;
            rhs = rec->data + rhs_col->offset;
        }
        assert(rhs_type == lhs_col->type);  // TODO convert to common type
        int cmp = ix_compare(lhs, rhs, rhs_type, lhs_col->len);
        if (cond.op == OP_EQ) {
            return cmp == 0;
        } else if (cond.op == OP_NE) {
            return cmp != 0;
        } else if (cond.op == OP_LT) {
            return cmp < 0;
        } else if (cond.op == OP_GT) {
            return cmp > 0;
        } else if (cond.op == OP_LE) {
            return cmp <= 0;
        } else if (cond.op == OP_GE) {
            return cmp >= 0;
        } else {
            throw InternalError("Unexpected op type");
        }
    }

    bool eval_conds(const std::vector<ColMeta> &rec_cols, const std::vector<Condition> &conds, const RmRecord *rec) {
        return std::all_of(conds.begin(), conds.end(),
                           [&](const Condition &cond) { return eval_cond(rec_cols, cond, rec); });
    }
};